package org.onion.pattern.event.springboot.rocketmq.consume.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.onion.pattern.event.consumer.EventConsumeHandler;
import org.onion.pattern.event.spring.io.EventCodecs;
import org.onion.pattern.event.springboot.rocketmq.consume.TopicAndEventTypeMapping;

import java.text.MessageFormat;
import java.util.List;

@Slf4j
public class RocketMessageConcurrentlyListener extends _MessageListener implements MessageListenerConcurrently {

    public RocketMessageConcurrentlyListener(DefaultMQPushConsumer consumer, EventConsumeHandler consumeHandler
            , EventCodecs eventCodecs, TopicAndEventTypeMapping topicAndEventTypeMapping) {
        this.consumeHandler = consumeHandler;
        this.consumer = consumer;
        this.topicAndEventTypeMapping = topicAndEventTypeMapping;
        this.eventCodecs = eventCodecs;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

        for (MessageExt messageExt : msgs) {
            try {
                log.debug(MessageFormat.format(
                        "【事件消费】接收到事件, group->{0}, topic->{1}, tags->{2}, message-id->{3}, message-key->{4}, queue->{5}",
                        consumer.getConsumerGroup(), messageExt.getTopic(), messageExt.getTags(),
                        messageExt.getMsgId(),
                        messageExt.getKeys(), messageExt.getQueueId()));

                handleMessage(messageExt);

//                if (subscribeResult.getExecutionStatus() == ExecutionStatus.FAILURE
//                        || subscribeResult.getExecutionStatus() == ExecutionStatus.UNKNOWN) {
//                    String pattern = "【事件消费】事件消费失败, group->{0}, topic->{1}, tag->{2}, message_key->{3}";
//                    Object[] args = {consumer.getConsumerGroup(), messageExt.getTopic(), messageExt.getTags(),
//                            messageExt.getKeys()};
//                    log.error(MessageFormat.format(pattern, args), subscribeResult.getException());
////                    log.warn("消费失败，consumer group：{}，topic：{}，tags:{}，key:{}，message：{}"
////                            , consumer.getConsumerGroup()
////                            , messageExt.getTopic()
////                            , messageExt.getTags()
////                            , messageExt.getKeys()
////                            , Optional.ofNullable(ExceptionUtils.getRootCause(subscribeResult.getException()))
////                                    .map(Throwable::getMessage)
////                                    .orElse("NO_ROOT_EX-" + subscribeResult.getException().getMessage()));
//
//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                }
            } catch (Throwable e) {
                log.error("【事件消费】失败, group->{}, topic->{}, tag->{}, message_key->{}",
                        consumer.getConsumerGroup(), messageExt.getTopic(), messageExt.getTags(),
                        messageExt.getKeys(), e);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }

        }

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }


}
